package com.world.rickcloudendpoint.demo.redis;

import com.world.rickcloudendpoint.common.ThreadNameFactory;
import com.world.rickcloudendpoint.demo.QueueConfig;
import com.world.rickcloudendpoint.demo.RedisService;
import com.world.rickcloudendpoint.demo.listener.RedisQueueListener;
import lombok.extern.log4j.Log4j2;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @description: 消息分发器
 * @author: jlsong
 * @date: 2022/03/17 21:38:53
 * @version: 1.0
 */
@Log4j2
public class RedisMqConsumerContainer {

    private RedisService redisService;

    private Map<String, QueueConfig> consumerMap = new HashMap<>(16);

    public static boolean run;
    private ExecutorService executorService;

    public RedisMqConsumerContainer(RedisService redisService) {
        this.redisService = redisService;
    }

    public void addConsumer(QueueConfig config) {
        if (consumerMap.containsKey(config.getQueueName())) {
            log.warn("Key : {}  已经存在， ", config.getQueueName());
            return;
        }
        if (config.getConsumer() == null) {
            log.warn("Key：{} consumer为空，无法对其进行配置", config.getQueueName());
            return;
        }
        consumerMap.put(config.getQueueName(), config);
    }

    public void destroy() {
        run = false;
        this.executorService.shutdown();

        log.info("Redis queue listener 正在退出....");
        while (!this.executorService.isTerminated()) {

        }
        log.info("Redis queue listener 退出");
    }

    public void init() {
        run = true;
        this.executorService = Executors.newCachedThreadPool(new ThreadNameFactory("Redis-mq-listener-"));
        consumerMap = Collections.unmodifiableMap(consumerMap);
        consumerMap.forEach((k, v) -> executorService.submit(new RedisQueueListener(redisService, v.getQueueName(), v.getConsumer())));
    }
}
